#include "SHMQueue.h"

namespace xncc {
namespace msg {
    using namespace xncc::shared;
    static constexpr int HEADER_OFFSET  = sizeof(shm_queue_index_t) * 6;
    static constexpr int INDEX_TYPE_LEN = sizeof(shm_queue_index_t);

    SHMQueue* SHMQueue::createProducerQueue(shm_queue_index_t size, void* mem) { return new (mem) SHMQueue(size, mem); }

    SHMQueue* SHMQueue::createConsumerQueue(void* mem) { return new (mem) SHMQueue(mem); }

    shm_queue_index_t SHMQueue::GetMemorySize(shm_queue_index_t size) { return HEADER_OFFSET + size; }

    void SHMQueue::destroyQueue(SHMQueue* queue) {}

    SHMQueue::SHMQueue(shm_queue_index_t size, void* mem)
        : MASK(size - HEADER_OFFSET - 1), out(0), currReadLen(0), in(0), currWriteLen(0), currentFinal(0)
    {
        if (!CommonBusi::is_power_of_2(size - HEADER_OFFSET)) {
            throw logic_exception("size error, must be power of 2");
        }
    }

    SHMQueue::SHMQueue(void* mem)
        : MASK(*reinterpret_cast<shm_queue_index_t*>(mem)),
          out(*(reinterpret_cast<shm_queue_index_t*>(mem) + INDEX_TYPE_LEN)),
          currReadLen(*(reinterpret_cast<shm_queue_index_t*>(mem) + INDEX_TYPE_LEN * 2)),
          in(*(reinterpret_cast<shm_queue_index_t*>(mem) + INDEX_TYPE_LEN * 3)),
          currWriteLen(*(reinterpret_cast<shm_queue_index_t*>(mem) + INDEX_TYPE_LEN * 4)),
          currentFinal(*(reinterpret_cast<shm_queue_index_t*>(mem) + INDEX_TYPE_LEN * 5))
    {
    }

    SHMQueue::~SHMQueue() {}

    char* SHMQueue::getWriteBuffer(shm_queue_index_t len)
    {
        const auto cached_out  = out;
        const auto cached_in   = in;
        const auto read_index  = static_cast<shm_queue_index_t>(cached_out & MASK);
        const auto write_index = static_cast<shm_queue_index_t>(cached_in & MASK);
        const auto unused      = (capacity() - static_cast<shm_queue_index_t>(cached_in - cached_out));
        // case1
        //                              w_index        r_index      Final
        //|-------------------------------|-------------|------------|---------------|
        //                                |<-  unused ->|            |               |
        // 如果写数据不超过unused 不影响Final 写数据超过unused 写入失败 不影响Final

        // case2
        //                              w_index        Final      r_index
        //|-------------------------------|-------------|------------|---------------|
        //                                |<-         unused       ->|               |
        // 这种情况不会出现 rindex 遇到final 就会回到起点

        // case3
        //                              r_index        w_index      Final
        //|-------------------------------|-------------|------------|---------------|
        //|<-             unused        ->|             |<-         unused         ->|
        // 右边的unused的已经read了 或者没有read（只发生在刚开始写数据的时候 Final是size）已经read 此次write
        // 会影响Final
        if (len > unused) {
            return nullptr;
        }
        if (static_cast<shm_queue_index_t>(capacity() - write_index) >= len) {
            currWriteLen = len;
            if (static_cast<shm_queue_index_t>(write_index + len) > currentFinal) {
                currentFinal = capacity();
            }
            return reinterpret_cast<char*>(this) + write_index;
        }
        else {
            if (len <= read_index) {
                currWriteLen = len + static_cast<shm_queue_index_t>(capacity() - write_index);
                currentFinal = write_index;
                return reinterpret_cast<char*>(this);
            }
            else {
                return nullptr;
            }
        }

        return nullptr;
    }

    void SHMQueue::finishWrite()
    {
        smp_wmb();
        const auto cached_in = in;
        const auto result    = static_cast<shm_queue_index_t>(cached_in + currWriteLen);
        in                   = result;
    }

    char* SHMQueue::getReadBuffer(shm_queue_index_t len)
    {
        const auto cached_in   = in;
        const auto cached_out  = out;
        const auto read_index  = static_cast<shm_queue_index_t>(cached_out & MASK);
        const auto write_index = static_cast<shm_queue_index_t>(cached_in & MASK);
        if (static_cast<shm_queue_index_t>(cached_in - cached_out) >= len) {
            __asm__ __volatile__("" ::: "memory");  // prevent new write_index old final
            if (read_index == currentFinal) {
                if (write_index >= len) {
                    currReadLen = len + static_cast<shm_queue_index_t>(capacity() - read_index);
                    return reinterpret_cast<char*>(this);
                }
                else {
                    fprintf(stderr, "read  what happened ? \n");
                }
            }
            else {
                currReadLen = len;
                return reinterpret_cast<char*>(this) + read_index;
            }
        }
        else {
            return nullptr;
        }
        return nullptr;
    }

    void SHMQueue::finishRead()
    {
        smp_rmb();
        const auto cached_out = out;
        const auto result     = static_cast<shm_queue_index_t>(cached_out + currReadLen);
        out                   = result;
    }

}  // namespace msg
}  // namespace xncc
